package cn.monkey.io.transport.mq.redisson;

import cn.monkey.io.transport.mq.ConnectionProperties;
import cn.monkey.io.transport.mq.MqInboundConfig;
import cn.monkey.io.transport.mq.MqOutboundConfig;
import cn.monkey.io.transport.core.Exchange;
import org.redisson.Redisson;
import org.redisson.api.RedissonClient;
import org.redisson.client.codec.Codec;
import org.redisson.config.Config;

public class SimpleRedissonExchangeFactory extends AbstractRedissonExchangeFactory<RedissonExchange> {

    protected RedissonClient redissonClient(ConnectionProperties connectionProperties) {
        String[] addresses = connectionProperties.getAddresses();
        Config config = new Config();
        if (addresses.length > 1) {
            config.useClusterServers()
                    .addNodeAddress(addresses);

        } else {
            config.useSingleServer().setAddress(addresses[0]);
        }
        return Redisson.create(config);
    }

    protected Codec codec(RedissonTransportConfig redissonTransportConfig) {
        return redissonTransportConfig.getCodec();
    }

    @Override
    protected RedissonInbound newInbound(String key, MqInboundConfig inbound, Codec codec) {
        String[] topics = inbound.getTopics();
        if (topics == null || topics.length == 0) {
            topics = new String[]{key + "_inbound"};
            inbound.setTopics(topics);
        }
        return new SimpleRedissonInbound(this.redissonClient(inbound), codec, inbound);
    }

    @Override
    protected RedissonOutbound newOutbound(String key, MqOutboundConfig outbound, Codec codec) {
        String[] topics = outbound.getTopics();
        if (topics == null || topics.length == 0) {
            topics = new String[]{key + "_outbound"};
            outbound.setTopics(topics);
        }
        return new SimpleRedissonOutbound(redissonClient(outbound), codec, outbound);
    }

    @Override
    public RedissonExchange apply(String key, Exchange exchange, RedissonTransportConfig redissonExchangeConfig) {
        if (exchange == null) {
            Codec codec = codec(redissonExchangeConfig);
            SimpleRedissonExchange redissonExchange = new SimpleRedissonExchange(key, this.newInbound(key, redissonExchangeConfig.getInbound(), codec), this.newOutbound(key, redissonExchangeConfig.getOutbound(), codec));
            return redissonExchange.bind();
        }
        if (exchange instanceof RedissonExchange redissonExchange) {
            return redissonExchange;
        }
        throw new IllegalArgumentException("exchange should be redissonExchange or null");
    }
}
